package io.github.wanggit.antrpc.client.zk.zknode;

import io.github.wanggit.antrpc.client.zk.register.ZkRegisterException;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
public class ZkNodeOperator implements IZkNodeOperator {

    private final CuratorFramework curator;

    public ZkNodeOperator(CuratorFramework curator) {
        this.curator = curator;
    }

    @Override
    public void remoteCreateZkNode(String zkFullpath, byte[] nodeData, CreateMode createMode) {
        // 因为zk的临时节点超时需要一段时间，如果在服务快速重启时，可能上一次会话的临时节点还未消失，所以在此手动删除一次
        try {
            curator.delete().deletingChildrenIfNeeded().forPath(zkFullpath);
        } catch (KeeperException.NoNodeException e) {
            if (log.isInfoEnabled()) {
                log.info(
                        "The "
                                + zkFullpath
                                + " node for the last session has been deleted. "
                                + e.getMessage());
            }
        } catch (KeeperException.NotEmptyException e) {
            if (log.isInfoEnabled()) {
                log.info(
                        "The "
                                + zkFullpath
                                + " node has children and cannot be deleted. "
                                + e.getMessage());
            }
        } catch (Exception e) {
            throw new ZkRegisterException(
                    "An exception occurred while deleting the "
                            + zkFullpath
                            + " node from the last session.",
                    e);
        }
        try {
            String forPath =
                    curator.create()
                            .creatingParentsIfNeeded()
                            .withMode(createMode)
                            .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                            .forPath(zkFullpath, nodeData);
            if (log.isInfoEnabled()) {
                log.info("Zookeeper node " + forPath + " was created successfully.");
            }
        } catch (Exception e) {
            throw new ZkRegisterException("Zookeeper node " + zkFullpath + " creation failed.", e);
        }
    }

    @Override
    public void deleteNode(String zookeeperFullPath) {
        try {
            curator.delete().deletingChildrenIfNeeded().forPath(zookeeperFullPath);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Failed to delete " + zookeeperFullPath + " node.", e);
            }
        }
    }

    @Override
    public boolean existsNode(String fullPath) {
        try {
            Stat stat = curator.checkExists().forPath(fullPath);
            return null != stat;
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("check node " + fullPath + " exists error.");
            }
        }
        return false;
    }

    @Override
    public boolean existsData(String fullPath) {
        byte[] bytes = null;
        try {
            bytes = curator.getData().forPath(fullPath);
        } catch (Exception e) {
            if (log.isInfoEnabled()) {
                log.info("check node " + fullPath + " has data error." + e.getMessage());
            }
        }
        return null != bytes;
    }

    @Override
    public byte[] getData(String fullPath) {
        try {
            return curator.getData().forPath(fullPath);
        } catch (Exception e) {
            if (log.isInfoEnabled()) {
                log.info("get node " + fullPath + " data error. ", e);
            }
        }
        return new byte[0];
    }

    @Override
    public void setNodeData(String fullPath, byte[] data) {
        try {
            curator.setData().forPath(fullPath, data);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("set node data " + fullPath + " error.", e);
            }
        }
    }

    @Override
    public List<String> getChildren(String path) {
        try {
            return curator.getChildren().forPath(path);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("get children " + path + " error.", e);
            }
        }
        return new ArrayList<>();
    }

    @Override
    public void getDescendants(String path, Filter filter, List<String> descendants) {
        List<String> result = null == descendants ? new ArrayList<>() : descendants;
        List<String> children = getChildren(path);
        children = children.stream().map(it -> path + "/" + it).collect(Collectors.toList());
        if (children.isEmpty() && filter.accepted(path)) {
            result.add(path);
        }
        children.forEach(it -> getDescendants(it, filter, result));
    }
}
